package net.lc7.curator;

import net.lc7.util.ZkPropertiesUtil;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.utils.EnsurePath;

/**
 * @Description: Leader选举
 * 当集群里的某个服务down机时，我们可能要从slave结点里选出一个作为新的master，这时就需要一套能在分布式环境中自动协调的Leader选举方法。
 * Curator提供了LeaderSelector监听器实现Leader选举功能。同一时刻，只有一个Listener会进入takeLeadership()方法，说明它是当前的Leader。
 * 注意：当Listener从takeLeadership()退出时就说明它放弃了“Leader身份”，这时Curator会利用Zookeeper再从剩余的Listener中选出一个新的Leader。
 * autoRequeue()方法使放弃Leadership的Listener有机会重新获得Leadership，如果不设置的话放弃了的Listener是不会再变成Leader的。
 * @Author: Jason.zhu
 * @Create: 2019/05/27 11:49
 */

public class CuratorLeaderClient {
    private static String path = "/ensurePath";

    public static void main(String[] args) throws InterruptedException {
        LeaderSelectorListener listener = new LeaderSelectorListener() {
            @Override
            public void takeLeadership(CuratorFramework curatorFramework) throws Exception {
                System.out.println(Thread.currentThread().getName() + " take leadership !!");
                Thread.sleep(5000L);
                System.out.println(Thread.currentThread().getName() + " relinquish leadership !!");
            }

            @Override
            public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
                System.out.println(connectionState.name() + " state changed !! " + connectionState.isConnected());
            }
        };

        new Thread(() -> {
            registerListener(listener);
        }).start();

        new Thread(() -> {
            registerListener(listener);
        }).start();


        new Thread(() -> {
            registerListener(listener);
        }).start();


        new Thread(() -> {
            registerListener(listener);
        }).start();

        new Thread(() -> {
            registerListener(listener);
        }).start();

        Thread.sleep(Integer.MAX_VALUE);
    }

    public static void registerListener(LeaderSelectorListener listener){
        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(ZkPropertiesUtil.getZkServerIp()
                                                                    ,new RetryNTimes(10, 5000));

        curatorFramework.start();

        //ensure path
        try {
            new EnsurePath(path).ensure(curatorFramework.getZookeeperClient());
        } catch (Exception e) {
            e.printStackTrace();
        }

        //register listener
        LeaderSelector selector = new LeaderSelector(curatorFramework, path, listener);
        selector.autoRequeue();
        selector.start();
    }

}
